package com.uzhie.media.service;

import com.alibaba.fastjson.JSONObject;
import com.uzhie.media.bean.MediaMsgBean;
import com.uzhie.media.task.CallableTask;
import com.uzhie.media.task.MediaOnLineNumTask;
import com.uzhie.util.CacheManager;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.concurrent.ExecutionException;

/**
 * Created by TangFei on 2018/7/28.
 */
public class MediaPostServiceImpl<CM> implements MediaPostService {

    private Logger logger = Logger.getLogger(getClass());

    @Autowired(required = false)
    private CacheManager<String, Channel> channelCacheManager;
    @Autowired(required = false)
    private CacheManager<String, CM> cacheManager;

    @Autowired
    @Qualifier(value = "mediaThreadExecutor")
    private ThreadPoolTaskExecutor taskExecutor;

    @Override
    public void post(int msgID, JSONObject body) {
        MediaMsgBean mediaMsgBean = new MediaMsgBean();
        mediaMsgBean.setMsgID(msgID);
        mediaMsgBean.setMsgBody(body);
        // 下发指令
        Channel channel = channelCacheManager.get(CacheManager.DES.MEDIA_SESSION_CACHE, "mediaTCPSession", Channel.class);
        logger.info("获取到流媒体连接通道：" + channel);
        ChannelFuture channelFuture = channel.writeAndFlush(mediaMsgBean);
        logger.info("result " + channelFuture.isSuccess());
    }

    /**
     * 发送请求 获取请求返回值
     *
     * @param msgID
     * @param body
     * @param resultClass
     * @param <T>
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @Override
    public <T> T postWithResult(int msgID, JSONObject body, Class<? extends CallableTask> taskClass, Class<T> resultClass) {
        T result = null;
        MediaMsgBean mediaMsgBean = new MediaMsgBean();
        mediaMsgBean.setMsgID(msgID);
        mediaMsgBean.setMsgBody(body);
        // 下发指令
        try {
            Channel channel = channelCacheManager.get(CacheManager.DES.MEDIA_SESSION_CACHE, "mediaTCPSession", Channel.class);
            ChannelFuture channelFuture = channel.writeAndFlush(mediaMsgBean);
            Constructor<? extends CallableTask> constructor = taskClass.getDeclaredConstructor(JSONObject.class, CacheManager.class);
            CallableTask instance = constructor.newInstance(body, cacheManager);
            // 获取响应结果
            java.util.concurrent.Future<T> future = taskExecutor.submit(instance);
            result = future.get();
        } catch (NoSuchMethodException e) {
            logger.error("找不到任务构造方法", e);
        } catch (InvocationTargetException e) {
            logger.error("任务类调用失败", e);
        } catch (NullPointerException e) {
            logger.error("车辆不在线", e);
        } finally {
            return null == result ? result : resultClass.cast(result);
        }
    }

    /**
     * 发送请求 获取请求返回值
     *
     * @param msgID
     * @param body
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public <V> V postWithResult(int msgID, JSONObject body) throws ExecutionException, InterruptedException {
        MediaMsgBean mediaMsgBean = new MediaMsgBean();
        mediaMsgBean.setMsgID(msgID);
        mediaMsgBean.setMsgBody(body);
        // 下发指令
        Channel channel = channelCacheManager.get(CacheManager.DES.MEDIA_SESSION_CACHE, "mediaTCPSession", Channel.class);
        ChannelFuture channelFuture = channel.writeAndFlush(mediaMsgBean);
        logger.info("result " + channelFuture.isSuccess());
        // 获取响应结果
        MediaOnLineNumTask mediaOnLineNumTask = new MediaOnLineNumTask(body, cacheManager);
        java.util.concurrent.Future<V> future = taskExecutor.submit(mediaOnLineNumTask);
        V result = future.get();
        return result;
    }

}
